package com.gitee.sparta.common;


import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.gitee.sparta.common.exception.SpartaException;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RScript;
import org.redisson.api.RedissonClient;

import java.lang.reflect.Method;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

/**
 * sparta 内部定义的 lua 脚本
 * <p>
 * 本类中所定义的脚本全部都是兼容 阿里云 redis cluster 集群限制的
 *
 * @author zyred
 * @since v 0.0.1
 */
@SuppressWarnings("all")
@Slf4j
public class SpartaLuaConstant {

    private static final SpartaLuaConstant INSTANCE = new SpartaLuaConstant();
    private static final Map<String, Map<String, String>> LUA_SCRIPT_CACHE = new HashMap<>();

    private SpartaLuaConstant() {
    }

    /**
     * 将 <method, sha> 解析成  <method, <sha, script>> 数据结构
     * @param shaTab
     */
    public static void parseLoadedSha(Map<String, String> shaTab) {
        Method[] scriptMethods = SpartaLuaConstant.class.getDeclaredMethods();
        try {
            for (Method scriptMethod : scriptMethods) {

                String sha = shaTab.get(scriptMethod.getName());
                if (StrUtil.isNotBlank(sha)) {
                    String lua = (String) scriptMethod.invoke(INSTANCE);
                    LUA_SCRIPT_CACHE.put(scriptMethod.getName(), createMap(lua, sha));
                }
            }
        } catch (Exception ex) {
            log.error("[sparta start failure] exception: {}", ex.getMessage(), ex);
        }
    }


    /**
     * 加载 lua script 到 redis 服务器中
     *
     * @param client redisson client
     */
    public static Map<String, String> loadScript(RedissonClient client) {
        /** method -> sha **/
        Map<String, String> methodSha = new HashMap<>(32);
        RScript script = client.getScript();
        Method[] declaredMethods = SpartaLuaConstant.class.getDeclaredMethods();
        try {
            for (Method declaredMethod : declaredMethods) {
                Class<?>[] parameterTypes = declaredMethod.getParameterTypes();
                // 如果返回结果是 string 那么说明该方法是返回的脚本，这是规则
                if (declaredMethod.getReturnType() == String.class && parameterTypes.length == 0) {

                    String methodName = declaredMethod.getName();
                    String luaScript = (String) declaredMethod.invoke(INSTANCE);
                    String sha = script.scriptLoad(luaScript);
                    methodSha.put(methodName, sha);
                    LUA_SCRIPT_CACHE.put(methodName, createMap(luaScript, sha));
                }
            }
            return methodSha;
        } catch (Exception ex) {
            log.error("[sparta start failure] exception: {}", ex.getMessage(), ex);
            return Collections.emptyMap();
        }
    }


    public static Map<String, String> getSha(String methodName) {
        Map<String, String> cache = LUA_SCRIPT_CACHE.get(methodName);
        if (CollUtil.isEmpty(cache)) {
            throw new SpartaException("in cache method name " + methodName + " not found..");
        }
        return cache;
    }


    private static Map<String, String> createMap(String script, String sha) {
        Map<String, String> shaMapping = new HashMap<>(2);
        shaMapping.put(SpartaConstant.LUA_SHA_KEY, sha);
        shaMapping.put(SpartaConstant.SCRIPT_KEY, script);
        return shaMapping;
    }

    /**
     * 广播消息写入
     *
     * @return  脚本
     */
    public static String getWriteBroadcastScript() {
        return "redis.call('ZAdd', KEYS[2], ARGV[3], ARGV[2])\n" +
                "redis.call('HSet', KEYS[1], ARGV[1], ARGV[4])";
    }

    /**
     * 分段消息写入
     *
     * @return  脚本
     */
    public static String getWriteSegmentScript() {
        return "local millis = {}\n" +
                "string.gsub(ARGV[3], '[^,]+', function(w)\n" +
                "    table.insert(millis, w)\n" +
                "end)\n" +
                "for idx, val in pairs(millis) do\n" +
                "    idx = idx - 1\n" +
                "    ARGV[5] = val\n" +
                "    ARGV[6] = ARGV[1] .. ARGV[4] .. idx\n" +
                "    redis.call('ZAdd', KEYS[2], ARGV[5], ARGV[6])\n" +
                "end\n" +
                "redis.call('HSet', KEYS[1], ARGV[1], ARGV[2])";
    }

    /**
     * 普通消息写入
     *
     * @return  脚本
     */
    public static String getWriteNormalScript() {
        return "redis.call('ZAdd', KEYS[2], ARGV[2], ARGV[1])\n" +
                "redis.call('HSet', KEYS[1], ARGV[1], ARGV[3])";
    }

    /**
     * 执行搬运的脚本
     * 1. 从 timeout_tab 中获取总数量，如果有内容的情况
     * 2. 如果 timeout_tab 中有内容，则根据 0 -> currentTimestamp 范围内取值
     * 3. 如果取不到值的情况，获取现在 timeout_tab 中最小的分数，该分数作为核心睡眠时间的计算因素，并构造结果进行返回
     * 4. 如果取到了值，遍历取到的值，每一个值都与当前时间做对比，必须要满足 >= 当前时间
     * 5. 处理广播的标识，如果 topic 中含有广播的标识，直接从 timeout_tab 中删除
     * 6. 如果不是广播，说明是队列，则需要将 timeout_tab 中的数据搬运到 readable_tab 中，并且删除 timeout_tab 中的内容
     * 7. 组装需要返回的数据，然后将内容 return
     *
     * @return [已经搬运的队列 topic，最小分数，广播的 topic]
     */
    public static String getCarryScript() {
        return "local resultMap = {}\n" +
                "local member = redis.call('ZRangeByScore', KEYS[2], 0, tonumber(ARGV[1]))\n" +
                "if next(member) == nil then\n" +
                "    local minScore = redis.call('ZRangeByScore', KEYS[2], '-inf', '+inf', 'WithScores', 'limit', 0, 1)\n" +
                "    table.insert(resultMap, 1, '')\n" +
                "    table.insert(resultMap, 2, minScore ~= nil and minScore[2] or '0')\n" +
                "    table.insert(resultMap, 3, '')\n" +
                "else\n" +
                "    local topics = {}\n" +
                "    local channels = ''\n" +
                "    for _, v in ipairs(member) do\n" +
                "        local needIf = string.find(v, ARGV[3])\n" +
                "        if needIf ~= nil then\n" +
                "            redis.call('ZRem', KEYS[2], v)\n" +
                "            local full = string.sub(v, 2, #v)\n" +
                "            channels = channels .. ',' .. full\n" +
                "        else\n" +
                "            local index = string.find(v, ':')\n" +
                "            local subTopic = string.sub(v, 1, index - 1)\n" +
                "            KEYS[4] = KEYS[1] ..':'.. ARGV[2] ..':'..subTopic\n" +
                "            redis.call('ZRem', KEYS[2], v)\n" +
                "            redis.call('SAdd', KEYS[4], v)\n" +
                "            KEYS[4] = nil\n" +
                "            topics[subTopic] = true\n" +
                "        end\n" +
                "    end\n" +
                "    local minScore = redis.call('ZRangeByScore', KEYS[2], '-inf', '+inf', 'WithScores', 'limit', 0, 1)\n" +
                "    local tp = ''\n" +
                "    for tk, _ in pairs(topics) do\n" +
                "        tp = tp .. ',' .. tk\n" +
                "    end\n" +
                "    table.insert(resultMap, 1, #tp ~= 0 and tp or '')\n" +
                "    table.insert(resultMap, 2, minScore ~= nil and minScore[2] or '0')\n" +
                "    table.insert(resultMap, 3, #channels ~= 0 and channels or '')\n" +
                "end\n" +
                "return resultMap";
    }

    /**
     * 读取 metadta_tab 中的数据脚本
     *
     * @return {@see Metadata}
     */
    public static String getReadMetadataScript() {
        return "return redis.call('HGet', KEYS[1], ARGV[1])";
    }

    /**
     * 删除 readable_tab 中已经被读取的数据
     *
     * @return str
     */
    public static String getDelHaveReadInReadableTabScript() {
        return "redis.call('SRem', KEYS[1], ARGV[1])";
    }

    /**
     * 删除 readable_tab 与 metadata_tab 中的数据
     *
     * @return void
     */
    public static String getDelMetadataReadableTabScript() {
        return "redis.call('SRem', KEYS[1], ARGV[1]);\n" +
                "redis.call('HDel', KEYS[2], ARGV[1]);";
    }

    /**
     * 1. 删除 readable_tab
     * 2. 添加到 timeout_tab 中
     * 3. 添加到 metadata_tab 中
     *
     * @return void
     */
    public static String getRenewAndRetryScript() {
        return "redis.call('ZAdd', KEYS[1], ARGV[1], ARGV[2]); \n" +
                "redis.call('HSet', KEYS[2], ARGV[2], ARGV[3]); \n" +
                "redis.call('SRem', KEYS[3], ARGV[3]); ";
    }


    /**
     * 1. 全量查询 deny_tab 数据
     * 2. 遍历结果
     * 3. 每一个从 metadata_tab 中获取内容
     * 4. 删除 deny_tab, 删除 metadata_tab
     *
     * @return [{metadata1}, {medatada2}]
     */
    public static String getProcessDenysScript() {
        return "local resultList = {};\n" +
                "local size = redis.call('SCard', KEYS[1]);\n" +
                "if (tonumber(size) > 0) then\n" +
                "    local denyTab = redis.call('SMembers', KEYS[1]);\n" +
                "    for k, v in ipairs(denyTab) do\n" +
                "        ARGV[1] = v;\n" +
                "        local metadata = redis.call('HGet', KEYS[2], ARGV[1]);\n" +
                "        table.insert(resultList, k, metadata);\n" +
                "        redis.call('HDel', KEYS[2], ARGV[1]);\n" +
                "        redis.call('SRem', KEYS[1], ARGV[1]);\n" +
                "        ARGV[1] = nil;\n" +
                "    end\n" +
                "    return resultList;\n" +
                "else\n" +
                "    return resultList;\n" +
                "end";
    }

    /**
     * 删除 readble_tab 中 fullTopic 的数据
     *
     * @return void
     */
    public static String getDelSegmentReadableTabScript() {
        return "redis.call('SRem', KEYS[1], ARGV[1]); ";
    }

    /**
     * 删除 readable_tab 与 metadata_tab
     *
     * @return void
     */
    public static String getDelSegmentFinishedConsumptionScript() {
        return "redis.call('SRem', KEYS[1], ARGV[1]);" +
                "redis.call('HDel', KEYS[2], ARGV[2]);";
    }

    /**
     * 删除 redis 队列中所有的内容
     *
     * @return  void
     */
    public static String getDeleteScript () {
        return "local segmentTopics = {}\n" +
                "string.gsub(ARGV[2], '[^,]+', function(w)\n" +
                "    table.insert(segmentTopics, w)\n" +
                "end)\n" +
                "if (#segmentTopics == 0) then\n" +
                "    redis.call('SRem', KEYS[1], ARGV[1])\n" +
                "    redis.call('ZRem', KEYS[2], ARGV[1])\n" +
                "    redis.call('HDel', KEYS[3], ARGV[1])\n" +
                "else\n" +
                "    for k, v in ipairs(segmentTopics) do\n" +
                "        ARGV[2] = v\n" +
                "        redis.call('SRem', KEYS[1], ARGV[2])\n" +
                "        redis.call('ZRem', KEYS[2], ARGV[2])\n" +
                "    end\n" +
                "    redis.call('HDel', KEYS[3], ARGV[1])\n" +
                "end";
    }

    /**
     * 1. 将 readable_tab 中所有的消息全部放回到 timeout_tab 中
     *
     * @return      selfTable1: 被删除的个数 \t selfTable2: 被删除的个数 .... 字符串
     */
    public static String getShutdownAfterRollbackScript () {
        return  "local readableTabKeys = {}\n" +
                "string.gsub(ARGV[2], '[^,]+', function(a)\n" +
                "    table.insert(readableTabKeys, a)\n" +
                "end)\n" +
                "for _, tabKey in ipairs(readableTabKeys) do\n" +
                "    KEYS[2] = tabKey\n" +
                "    local rollbackTopics = redis.call('SMembers', KEYS[2])\n" +
                "    for _, topic in ipairs(rollbackTopics) do\n" +
                "        ARGV[4] = topic\n" +
                "        redis.call('ZAdd', KEYS[1], ARGV[1], ARGV[4])\n" +
                "    end\n" +
                "    redis.call('del', KEYS[2])\n" +
                "end";
    }

    /**
     * 读取 redis 中 metadata_tab 的指定数据
     *
     * @return  [metadata1, metadata2, ..]
     */
    public static String getReadMetadataMapScript () {
        return  "local members = redis.call('SMembers', KEYS[1])\n" +
                "local result = {}\n" +
                "if members ~= nil then\n" +
                "    local topics = ''\n" +
                "    local tab = {}\n" +
                "    for _, v in pairs(members) do\n" +
                "        local line, _ = string.find(v, ARGV[1])\n" +
                "        topics = topics .. ',' .. v\n" +
                "        if line ~= nil then\n" +
                "            v = string.sub(v, 1, line -1)\n" +
                "        end\n" +
                "        table.insert(tab, v)\n" +
                "    end\n" +
                "    table.insert(result, 1, topics)\n" +
                "    table.insert(result, 2, redis.call('HMGet', KEYS[2], unpack(tab)))\n" +
                "end\n" +
                "return result";
    }


    /**
     * 读取 metadta_tab 中的数据并删除脚本
     *
     * @return {@see Metadata}
     */
    public static String getReadDelMetadataScript() {
        return "local metadata = redis.call('HGet', KEYS[1], ARGV[1]);\n" +
                "redis.call('HDel', KEYS[1], ARGV[1]);\n" +
                "return metadata;";
    }


    /**
     * 1. 删除 readable_tab
     * 2. 添加 deny_tab
     *
     * @return void
     */
    public static String getDenyPersistenceScript() {
        return "redis.call('SAdd', KEYS[1], ARGV[1]); return redis.call('SRem', KEYS[2], ARGV[2]);";
    }
}
